package com.qupeng.demo.kafka.kafkaspringbootstreams.common.kafka.streams;

import com.qupeng.demo.kafka.kafkaspringbootstreams.common.utils.TimeUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.streams.processor.TimestampExtractor;

import java.text.ParseException;

public class CreatingTimestampExtractor implements TimestampExtractor {
 
    @Override
    public long extract(ConsumerRecord<Object, Object> record, long partitionTime) {
        try {
            String eventTime = ((EventWithTimestamp)record.value()).getTimestamp();
            System.out.println("Event in value: " + eventTime + "     " + TimeUtils.parse(eventTime));
            System.out.println("Logging time by producer: " + TimeUtils.format(record.timestamp()) + "     " + record.timestamp());
            return TimeUtils.parse(eventTime);
        } catch (ParseException e) {
            return record.timestamp();
        }
    }

    public static interface EventWithTimestamp {
        String getTimestamp();
    }
}